package org.apache.flink.formats.jsontm;

import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.format.DecodingFormat;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.utils.DataTypeUtils;
import org.apache.flink.types.RowKind;

import java.io.Serializable;
import java.util.*;
import java.util.stream.Collectors;

/**
 * <ol>
 * date:2022/1/4 editor:yanghongjian
 * <li>创建文档</li>
 * <li>支持发出 INSERT</li>
 * </ol>
 * <ol>
 *
 * @author <a href="mailto:12719889@qq.com">YangHongJian</a>
 * @version 3.0
 * @since 1.8
 */
public class JsonTmDecodingFormat implements DecodingFormat<DeserializationSchema<RowData>> {
    public static final String VALUE_METADATA_PREFIX = "value.";

    private final ReadableConfig options;
    private List<String> metadataKeys;
    private final List<ReadableMetadata> metadatas;

    public JsonTmDecodingFormat(ReadableConfig options) {
        this.options = options;
        this.metadataKeys = Collections.emptyList();
        this.metadatas = new ArrayList<>();
    }

    @Override
    public void applyReadableMetadata(List<String> metadataKeys) {
        this.metadataKeys = metadataKeys;
    }

    @Override
    public Map<String, DataType> listReadableMetadata() {
        final Map<String, DataType> metadataMap = new LinkedHashMap<>();

        this.metadatas.forEach(m -> metadataMap.put(m.key, m.dataType));
        return metadataMap;
    }

    @Override
    public DeserializationSchema<RowData> createRuntimeDecoder(
            DynamicTableSource.Context context, DataType physicalDataType) {
        final List<ReadableMetadata> readableMetadataList = metadataKeys.stream()
                .map(k -> metadatas.stream()
                        .filter(rm -> rm.key.equals(k)).findFirst().orElse(null))
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
        final List<DataTypes.Field> metadataFieldList = readableMetadataList.stream()
                .map(m -> DataTypes.FIELD(m.key, m.dataType))
                .collect(Collectors.toList());
        final DataType producedDataType = DataTypeUtils.appendRowFields(physicalDataType, metadataFieldList);

        final TypeInformation<RowData> producedTypeInfo = context.createTypeInformation(producedDataType);

        return new JsonTmDeserializationSchema(this.options,
                physicalDataType, readableMetadataList, producedTypeInfo);
    }

    @Override
    public ChangelogMode getChangelogMode() {
        return ChangelogMode.newBuilder().addContainedKind(RowKind.INSERT).build();
    }


    /**
     * 在DataApiDynamicTableSourceFactory->createDynamicTableSource()中调用
     */
    public void addReadableMetadata(List<Column.MetadataColumn> metadataColumns) {
        metadataColumns.stream()
                .filter(k -> k.getMetadataKey().orElse(k.getName()).startsWith(VALUE_METADATA_PREFIX))
                .forEach(k -> {
                    String name = k.getMetadataKey().orElse(k.getName()).substring(VALUE_METADATA_PREFIX.length());
                    this.metadatas.add(new ReadableMetadata(name, k.getDataType()));
                });
    }


    /**
     * metadata列表通过format读取
     */
    static class ReadableMetadata implements Serializable {
        private static final long serialVersionUID = 1L;
        final String key;
        final DataType dataType;

        ReadableMetadata(String key, DataType dataType) {
            this.key = key;
            this.dataType = dataType;
        }
    }
}
